package com.caits.lbs.framework.services.redis;

import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.logging.log4j.LogManager;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisZSetCommands;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.SerializationException;
import org.springframework.util.ObjectUtils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.parser.Feature;
import com.caits.lbs.framework.utils.SpringUtils;
import com.caits.lbs.framework.utils.StringUtils;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;

import com.caits.lbs.framework.services.AppCommandProcess;

/**
 * <p>
 * RedisCacheHelperImpl
 * </p>
 * <p>
 * 通用redis操作类，序列化json方式
 * </p>
 *
 * @author 周华彬(zhouhuabin@ctfo.com, zhou_hua_bin@163.com)
 * @version 0.0.0
 *          <table style="border:1px solid gray;">
 *          <tr>
 *          <th width="100px">版本号</th>
 *          <th width="100px">动作</th>
 *          <th width="100px">修改人</th>
 *          <th width="100px">修改时间</th>
 *          </tr>
 *          <!-- 以 Table 方式书写修改历史 -->
 *          <tr>
 *          <td>0.0.0</td>
 *          <td>创建类</td>
 *          <td>Administrator</td>
 *          <td>2019年11月13日 上午11:16:18</td>
 *          </tr>
 *          <tr>
 *          <td>XXX</td>
 *          <td>XXX</td>
 *          <td>XXX</td>
 *          <td>XXX</td>
 *          </tr>
 *          </table>
 */
public class RedisDAOImpl implements RedisDAO, MessageListener {
	/** 变量:TODO,类型:Logger */
	protected final org.apache.logging.log4j.Logger LOG = LogManager.getLogger(getClass());
	private static final String LOCK_SUCCESS = "OK";
	private static final String SET_IF_NOT_EXIST = "NX";
	private static final String SET_WITH_EXPIRE_TIME = "EX";
	public static final String CHARSET_UTF8 = "utf-8";

	private RedisTemplate<String, Object> redisTemplate;

	 /** 变量:消息接收处理类,类型:xxx */
    private AppCommandProcess msgService;
    
	public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate) {
		this.redisTemplate = redisTemplate;
	}

	public RedisTemplate getRedisTemplate() {
		return redisTemplate;
	}
	/**
     * 获取属性 msgService的值
     * 
     * @return the msgService : ProcessJT703BDataService
     */
    public AppCommandProcess getMsgService() {
	if (msgService == null) {
	    try {
		msgService = SpringUtils.getApplicationContext().getBean(AppCommandProcess.class);
	    } catch (Exception e) {
		LOG.error("获取消息服务处理类异常,msg:" + e.getLocalizedMessage());
		return null;
	    }
	    if (msgService == null) {
		LOG.error("获取msgService服务为空");
		return null;
	    }
	}
		return msgService;
    }

    /**
     * 设置属性msgService的值，参数类型ProcessJT703BDataService
     * 
     * @param msgService
     *            the msgService to set
     */
    public void setMsgService(AppCommandProcess msgService) {
	this.msgService = msgService;
    }

	@Override
	public void sendMessage(String channels, Serializable message) {
		if (channels.contains(",")) {
			String[] chanArr = channels.split(",");
			for (String channel : chanArr) {
				try {
					redisTemplate.convertAndSend(channel, message);
				} catch (Exception e) {
					LOG.error("给redis发送消息错误,channel:" + channel + ",msg:" + e.getLocalizedMessage());
				}
			}
		} else {
			try {
				redisTemplate.convertAndSend(channels, message);
			} catch (Exception e) {
				LOG.error("给redis发送消息错误,channel:" + channels + ",msg:" + e.getLocalizedMessage());
			}
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.springframework.data.redis.connection.MessageListener#onMessage(org
	 * .springframework.data.redis.connection.Message, byte[])
	 */
	@Override
	public void onMessage(Message message, byte[] pattern) {
		// 什么都不做,只输出
		if (message == null) {
			LOG.warn("从redis中获取到消息null");
		} else {
			try {
				String body = null;
				try {
					body = (String) redisTemplate.getValueSerializer().deserialize(message.getBody());
				} catch (SerializationException e) {
					LOG.error("处理redis消息异常,尝试字符串格式,msg:" + e.getLocalizedMessage());
					body = new String(message.getBody());
				}
				String channel = (String) redisTemplate.getStringSerializer().deserialize(message.getChannel());
				LOG.info("从redis中获取到消息," + body);
				if (StringUtils.notNullOrBlank(body)) {
					getMsgService().processCommand(body); // 处理终端命令
				}
			} catch (Exception e) {
				LOG.error("处理redis消息异常,msg:" + e.getLocalizedMessage());
			}
		}
	}
	 @Override
		public long lpush(final String key, Object obj) {
			long result = redisTemplate.execute(new RedisCallback<Long>() {
				@Override
				public Long doInRedis(RedisConnection connection) throws DataAccessException {
					long count = 0;
					try {
						count = connection.lPush(key.getBytes(CHARSET_UTF8), toJson(obj));
					} catch (UnsupportedEncodingException e) {
						e.printStackTrace();
					}
					return count;
				}
			});
			return result;
		}

		@Override
		public long rpush(final String key, Object obj) {
			long result = redisTemplate.execute(new RedisCallback<Long>() {
				@Override
				public Long doInRedis(RedisConnection connection) throws DataAccessException {
					long count = 0;
					try {
						count = connection.rPush(key.getBytes(CHARSET_UTF8), toJson(obj));
					} catch (UnsupportedEncodingException e) {
						e.printStackTrace();
					}
					return count;
				}
			});
			return result;
		}

		@Override
		public String lpop(final String key) {
			String result = redisTemplate.execute(new RedisCallback<String>() {
				@Override
				public String doInRedis(RedisConnection connection) throws DataAccessException {
					byte[] res;
					try {
						res = connection.lPop(key.getBytes(CHARSET_UTF8));
						if (res != null) {
							return new String(res,CHARSET_UTF8);
						}
					} catch (UnsupportedEncodingException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
					return null;
				}
			});
			return result;
		}

		public String bLPop(final int timeOut, final String key) {
			String result = redisTemplate.execute(new RedisCallback<String>() {
				@Override
				public String doInRedis(RedisConnection connection) throws DataAccessException {
					List<byte[]> res=null;
					try {
						res = connection.bLPop(timeOut, key.getBytes(CHARSET_UTF8));
						if (res != null) {
							return new String(res.get(1),CHARSET_UTF8);
						}
					} catch (UnsupportedEncodingException e) {
						e.printStackTrace();
					}
					return null;
				}
			});
			return result;
		}

		public long ltrim(final String key, int start, int stop) {
			long result = redisTemplate.execute(new RedisCallback<Long>() {
				@Override
				public Long doInRedis(RedisConnection connection) throws DataAccessException {
					try {
						connection.lTrim(key.getBytes(CHARSET_UTF8), start, stop);
					} catch (UnsupportedEncodingException e) {
						e.printStackTrace();
					}
					return 1l;
				}
			});
			return result;
		}

		public <T> List<T> lrange(final String key, int start, int stop, Class<T> clazz) {
			List<T> result = redisTemplate.execute(new RedisCallback<List<T>>() {
				@Override
				public List<T> doInRedis(RedisConnection connection) throws DataAccessException {
					List<byte[]> dataList = new ArrayList<>(0);
					try {
						dataList = connection.lRange(key.getBytes(CHARSET_UTF8), start, stop);
					} catch (UnsupportedEncodingException e) {
						e.printStackTrace();
					}
					List<T> resultList = new ArrayList<>(dataList.size());
					for (byte[] data : dataList) {
						resultList.add(parseObject(data, clazz));
					}
					return resultList;
				}
			});
			return result;
		}

		public long llen(final String key) {
			long result = redisTemplate.execute(new RedisCallback<Long>() {
				@Override
				public Long doInRedis(RedisConnection connection) throws DataAccessException {
					long count = 0;
					try {
						count = connection.lLen(key.getBytes(CHARSET_UTF8));
					} catch (UnsupportedEncodingException e) {
						e.printStackTrace();
					}
					return count;
				}
			});
			return result;
		}
	public Boolean exists(final String key) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
				try {
					return connection.exists(key.getBytes(CHARSET_UTF8));
				} catch (UnsupportedEncodingException e) {
					e.printStackTrace();
				}
				return null;
			}
		});
	}

	public Boolean setnx(final String key, final String value) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			public Boolean doInRedis(RedisConnection redisConnection) throws DataAccessException {
				try {
					return redisConnection.setNX(key.getBytes(CHARSET_UTF8), value.getBytes(CHARSET_UTF8));
				} catch (UnsupportedEncodingException e) {
					e.printStackTrace();
				}
				return null;
			}
		});
	}
	/**
     * 写入分布式锁的值
     * @param key 锁
     * @param value 写入的值
     * @param expireTime 超期时间
     * @return 是否获取成功
     */
    public  Boolean setnx(final String key, final String value, int expireTime) {
    	/*
    	RedisConnection redisConn = redisTemplate.getConnectionFactory().getConnection();
    	JedisConnection jedisConn = (JedisConnection) redisConn;
    	Jedis jedis = jedisConn.getNativeConnection();
        String result = jedis.set(key, value, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
        if (LOCK_SUCCESS.equals(result)) {
            return true;
        }
        */
    	return redisTemplate.execute(new RedisCallback<Boolean>() {
    		 @Override
    		 public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
	    		  Object nativeConnection = connection.getNativeConnection();
	    		  String retStatusCode = "";
	    		  // 集群模式
	    		  if(nativeConnection instanceof JedisCluster) {
		    		   JedisCluster jedisCluster = (JedisCluster) nativeConnection;
		    		   retStatusCode = jedisCluster.set(key, value, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
	    		  }// 单机模式
	    		  else if(nativeConnection instanceof Jedis) {
		    		   Jedis jedis = (Jedis) nativeConnection;
		    		   retStatusCode = jedis.set(key, value, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
	    		  }
	    		  return ObjectUtils.nullSafeEquals(LOCK_SUCCESS, retStatusCode);
    		 }
    	});
    }
	@Override
	public String getString(final String key) {
		return redisTemplate.execute(new RedisCallback<String>() {
			@Override
			public String doInRedis(RedisConnection redisConnection) throws DataAccessException {
				try {
					byte[] res = redisConnection.get(key.getBytes(CHARSET_UTF8));
					if (res != null)
						return new String(res, CHARSET_UTF8);
				} catch (UnsupportedEncodingException e) {
					e.printStackTrace();
				}
				return null;
			}
		});
	}

	public Boolean expire(final String key, final long seconds) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
				try {
					if (connection.exists(key.getBytes(CHARSET_UTF8)))
						return connection.expire(key.getBytes(CHARSET_UTF8), seconds);
				} catch (UnsupportedEncodingException e) {
					e.printStackTrace();
				}
				return false;
			}
		});
	}

	public Long getExpireTime(final String key) {
		return (Long) redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				Long time = null;
				try {
					time = connection.ttl(key.getBytes(CHARSET_UTF8));
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				if (time == null) {
					return 0L;
				}
				return time;
			}
		});
	}

	public Long delCache(final String key) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				try {
					return connection.del(key.getBytes(CHARSET_UTF8));
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public Long incr(final String key, final Long n) {
		return (Long) redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				Long lreturn = null;
				try {
					lreturn = connection.incrBy(key.getBytes(CHARSET_UTF8), n);
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return lreturn;
			}
		});
	}

	public void addCache(final String key, final Object obj, final long seconds) {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				try {
					if (seconds <= 0) {
						connection.set(key.getBytes(CHARSET_UTF8), toJson(obj));
					} else {
						connection.setEx(key.getBytes(CHARSET_UTF8), seconds, toJson(obj));
					}
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public <T> T getCache(final String key, final Class<T> clazz) {
		return redisTemplate.execute(new RedisCallback<T>() {
			public T doInRedis(RedisConnection connection) throws DataAccessException {
				byte[] obj = new byte[0];
				try {
					obj = connection.get(key.getBytes(CHARSET_UTF8));
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				if (obj == null) {
					return null;
				}
				return parseObject(obj, clazz);
			}
		});
	}

	public <T> T getCache(final String key, final Type clazz) {
		return redisTemplate.execute(new RedisCallback<T>() {
			public T doInRedis(RedisConnection connection) throws DataAccessException {
				byte[] obj = new byte[0];
				try {
					obj = connection.get(key.getBytes(CHARSET_UTF8));
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				if (obj == null) {
					return null;
				}
				return parseObject(obj, clazz);
			}
		});
	}

	public <V> void hAdd(final String key, final Map<String, V> map, final long seconds) {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
				Map<byte[], byte[]> tmp = new HashMap<byte[], byte[]>();
				try {
					for (Map.Entry<String, V> entry : map.entrySet()) {
						tmp.put(entry.getKey().getBytes(CHARSET_UTF8), toJson(entry.getValue()));
					}
					redisConnection.hMSet(key.getBytes(CHARSET_UTF8), tmp);
					if (seconds > 0)
						redisConnection.expire(key.getBytes(CHARSET_UTF8), seconds);
				} catch (Exception e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public <V> Map<String, V> hGetAll(final String key, final Class<V> vClass) {
		return redisTemplate.execute(new RedisCallback<Map<String, V>>() {
			public Map<String, V> doInRedis(RedisConnection redisConnection) throws DataAccessException {
				Map<byte[], byte[]> tmp = null;
				try {
					tmp = redisConnection.hGetAll(key.getBytes(CHARSET_UTF8));
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				Map<String, V> map = new HashMap<String, V>();
				for (Map.Entry<byte[], byte[]> entry : tmp.entrySet()) {
					map.put(new String(entry.getKey()), parseObject(entry.getValue(), vClass));
				}
				return map;
			}
		});
	}

	public <T> T hGet(final String key, final String key2, final Class<T> clazz) {
		return redisTemplate.execute(new RedisCallback<T>() {
			public T doInRedis(RedisConnection connection) throws DataAccessException {
				byte[] obj = new byte[0];
				try {
					obj = connection.hGet(key.getBytes(CHARSET_UTF8), key2.getBytes(CHARSET_UTF8));
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				if (null == obj) {
					return null;
				} else {
					return parseObject(obj, clazz);
				}
			}
		});
	}

	public void hSet(final String key, final String key2, final Object obj) {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				try {
					connection.hSet(key.getBytes(CHARSET_UTF8), key2.getBytes(CHARSET_UTF8), toJson(obj));
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public Long hIncrBy(final String key, final String key2, final long incr) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				try {
					return connection.hIncrBy(key.getBytes(CHARSET_UTF8), key2.getBytes(CHARSET_UTF8), incr);
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public void hDel(final String key, final String key2) {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				try {
					connection.hDel(key.getBytes(CHARSET_UTF8), key2.getBytes(CHARSET_UTF8));
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public Boolean hExists(final String key, final String key2) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
				try {
					return connection.hExists(key.getBytes(CHARSET_UTF8), key2.getBytes(CHARSET_UTF8));
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public <T> void addListCache(final String key, final List<T> lists, final Long seconds) {
		redisTemplate.execute(new RedisCallback<T>() {
			public T doInRedis(RedisConnection connection) throws DataAccessException {
				if (lists == null || lists.size() == 0) {
					return null;
				}
				byte[][] bytes = new byte[lists.size()][];
				for (int i = 0; i < lists.size(); i++) {
					byte[] objByte = toJson(lists.get(i));
					bytes[i] = objByte;
				}
				try {
					connection.lPush(key.getBytes(CHARSET_UTF8), bytes);
					if (seconds > 0) {
						connection.expire(key.getBytes(CHARSET_UTF8), seconds);
					}
				} catch (Exception e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public <E> List<E> getListCache(final String key, final Class<E> clazz) {
		return redisTemplate.execute(new RedisCallback<List<E>>() {
			public List<E> doInRedis(final RedisConnection connection) throws DataAccessException {
				List<byte[]> obj = null;
				try {
					obj = connection.lRange(key.getBytes(CHARSET_UTF8), 0, connection.lLen(key.getBytes(CHARSET_UTF8)) - 1);
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				if (obj == null) {
					return null;
				}
				List<E> list = new ArrayList<E>();
				for (int i = 0; i < obj.size(); i++) {
					E o = parseObject(obj.get(i), clazz);
					list.add(o);
				}
				return list;
			}
		});
	}

	public void listTrim(final String key, final int start, final int stop) {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
				try {
					redisConnection.lTrim(key.getBytes(CHARSET_UTF8), start, stop);
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public void addSetCache(final String key, final Set<?> sets, final Long seconds) {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				if (sets == null || sets.size() == 0) {
					return null;
				}
				byte[][] bytes = new byte[sets.size()][];
				Iterator<?> iter = sets.iterator();
				int i = 0;
				while (iter.hasNext()) {
					byte[] objByte = toJson(iter.next());
					bytes[i] = objByte;
					i++;
				}
				try {
					connection.sAdd(key.getBytes(CHARSET_UTF8), bytes);
					if (seconds > 0) {
						connection.expire(key.getBytes(CHARSET_UTF8), seconds);
					}
				} catch (Exception e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public <T> Set<T> getSetCache(final String key, final Class<T> clazz) {
		return redisTemplate.execute(new RedisCallback<Set<T>>() {
			public Set<T> doInRedis(final RedisConnection connection) throws DataAccessException {
				Set<byte[]> obj = null;
				try {
					obj = connection.sMembers(key.getBytes(CHARSET_UTF8));
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				if (obj == null) {
					return null;
				}
				Set<T> set = new HashSet<T>();
				Iterator<byte[]> iter = obj.iterator();
				while (iter.hasNext()) {
					T o = parseObject(iter.next(), clazz);
					set.add(o);
				}
				return set;
			}
		});
	}

	public void sAdd(final String key, final Object value, final Long seconds) {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				if (value == null) {
					return null;
				}
				byte[] objByte = toJson(value);
				try {
					connection.sAdd(key.getBytes(CHARSET_UTF8), objByte);
					connection.expire(key.getBytes(CHARSET_UTF8), seconds);
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public Long sRem(final String key, final Object value) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				if (value == null) {
					return null;
				}
				byte[] objByte = toJson(value);
				try {
					return connection.sRem(key.getBytes(CHARSET_UTF8), objByte);
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public void zAdd(final String key, final Object value, final Long time) {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				if (value == null) {
					return null;
				}
				try {
					connection.zAdd(key.getBytes(CHARSET_UTF8), 1, toJson(value));
					connection.expire(key.getBytes(CHARSET_UTF8), time);
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public Long zRem(final String key, final Object value) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				if (value == null) {
					return null;
				}
				try {
					return connection.zRem(key.getBytes(CHARSET_UTF8), toJson(value));
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public <T> Set<T> zRange(final String key, final int from, final int to, final Class<T> clazz) {
		return redisTemplate.execute(new RedisCallback<Set<T>>() {
			public Set<T> doInRedis(RedisConnection connection) throws DataAccessException {
				Set<byte[]> zRange = null;
				try {
					zRange = connection.zRange(key.getBytes(CHARSET_UTF8), from, to);
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				HashSet<T> ret = new HashSet<T>();
				for (byte[] obj : zRange) {
					T t = parseObject(obj, clazz);
					ret.add(t);
				}
				return ret;
			}
		});
	}

	public void zRemSet(final String key, final Set<?> set) {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				if (set == null || set.size() == 0) {
					return null;
				}
				for (Object o : set) {
					byte[] objByte = toJson(o);
					try {
						connection.zRem(key.getBytes(CHARSET_UTF8), objByte);
					} catch (UnsupportedEncodingException e) {
						 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
					}
				}
				return null;
			}
		});
	}

	public Long zCard(final String key) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				try {
					return connection.zCard(key.getBytes(CHARSET_UTF8));
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public Double zIncrBy(final String key, final double score, final String member) {
		return redisTemplate.execute(new RedisCallback<Double>() {
			public Double doInRedis(RedisConnection connection) throws DataAccessException {
				try {
					return connection.zIncrBy(key.getBytes(CHARSET_UTF8), score, member.getBytes(CHARSET_UTF8));
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public Long zRemRangeByRank(final String key, final long start, final long end) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				try {
					return connection.zRemRange(key.getBytes(CHARSET_UTF8), start, end);
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public Map<String, Double> zRangeWithScores(final String key, final long start, final long end) {
		return redisTemplate.execute(new RedisCallback<Map<String, Double>>() {
			public Map<String, Double> doInRedis(RedisConnection connection) throws DataAccessException {
				Set<RedisZSetCommands.Tuple> res = null;
				try {
					res = connection.zRangeWithScores(key.getBytes(CHARSET_UTF8), start, end);
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				Map<String, Double> map = new HashMap<String, Double>();
				if (res != null && res.size() > 0) {
					for (RedisZSetCommands.Tuple tuple : res) {
						map.put(new String(tuple.getValue()), tuple.getScore());
					}
				}
				return map;
			}
		});
	}

	public <T> void addListCacheRight(final String key, final List<T> lists, final long seconds) {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
				if (lists == null || lists.size() == 0) {
					return null;
				}
				byte[][] bytes = new byte[lists.size()][];
				for (int i = 0; i < lists.size(); i++) {
					byte[] objByte = toJson(lists.get(i));
					bytes[i] = objByte;
				}
				try {
					redisConnection.rPush(key.getBytes(CHARSET_UTF8), bytes);
					if (seconds > 0) {
						redisConnection.expire(key.getBytes(CHARSET_UTF8), seconds);
					}
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public <T> void addListCacheRight(final String key, final T t, final long seconds) {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
				try {
					redisConnection.rPush(key.getBytes(CHARSET_UTF8), toJson(t));
					if (seconds > 0) {
						redisConnection.expire(key.getBytes(CHARSET_UTF8), seconds);
					}
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public void watch(final String key) {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
				try {
					redisConnection.watch(key.getBytes(CHARSET_UTF8));
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public void unwatch() {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
				redisConnection.unwatch();
				return null;
			}
		});
	}

	public <T> void delListCache(final String key, final T value) {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
				try {
					redisConnection.lRem(key.getBytes(CHARSET_UTF8), 0, toJson(value));
				} catch (UnsupportedEncodingException e) {
					 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
				}
				return null;
			}
		});
	}

	public void multi() {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
				redisConnection.multi();
				return null;
			}
		});
	}

	public void exec() {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
				redisConnection.exec();
				return null;
			}
		});
	}

	public void discard() {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
				redisConnection.discard();
				return null;
			}
		});
	}

	private byte[] toJson(Object object) {
		try {
			return JSON.toJSONString(object).getBytes(CHARSET_UTF8);
		} catch (UnsupportedEncodingException e) {
			 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
		}
		return null;
	}

	private <T> T parseObject(byte[] bytes, Class<T> clazz) {
		try {
			return JSON.parseObject(new String(bytes, CHARSET_UTF8), clazz);
		} catch (UnsupportedEncodingException e) {
			 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
		}
		return null;
	}

	private <T> T parseObject(byte[] bytes, Type type) {
		try {
			return JSON.parseObject(new String(bytes, CHARSET_UTF8), type, Feature.IgnoreNotMatch);
		} catch (UnsupportedEncodingException e) {
			 LOG.error("redis操作编码错误,msg:"+e.getLocalizedMessage());
		}
		return null;
	}
}
